package com.wangcanxuan.www.util;

import com.wangcanxuan.www.exception.RejectConnectionException;
import com.wangcanxuan.www.util.pool.PooledConnection;
import com.wangcanxuan.www.util.pool.factory.CheckConnectionThreadFactory;
import com.wangcanxuan.www.util.pool.handler.RejectedConnectionHandler;
import com.wangcanxuan.www.util.pool.total.ConnectionsTotal;

import java.sql.*;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicIntegerArray;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Logger;

import static com.wangcanxuan.www.util.constant.PooledConstants.*;

/**
 * 连接池跑龙套
 * 连接池工具类,使用单例模式，确保只有一个数据库连接池
 *
 * @author HHLJ
 * @date 2022/03/16
 */
public class ConnectionPoolUtil {

    static {
        try {
            Class.forName(DRIVER);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }

    /**
     * 连接池对象，单例
     */
    private static volatile ConnectionPoolUtil connectionPool = null;
    /**
     * 检查连接线程池
     */
    private static ScheduledExecutorService checkConnectionPool;
    /**
     * 存储连接的集合
     */
    private static List<PooledConnection> connections;
    /**
     * 数组阻塞队列，用来存储排队的连接
     */
    private ArrayBlockingQueue<PooledConnection> connectionsBlockingQueue;
    /**
     * 标志位数组
     */
    AtomicIntegerArray sign;
    /**
     * 写锁
     */
    private static final Lock WRITE_LOCK = new ReentrantReadWriteLock().writeLock();
    /**
     * 日志记录器
     */
    private static final Logger logger = LoggerUtil.getLogger();
    /**
     * 拒绝策略
     */
    private static RejectedConnectionHandler handler;

    /**
     * 本地线程，确保事务在一个线程处理
     */
    private static final ThreadLocal<PooledConnection> THREAD_LOCAL = new ThreadLocal<PooledConnection>() {
        @Override
        protected PooledConnection initialValue() {
            PooledConnection pooledConnection = connectionPool.getConnection();
            if (pooledConnection != null) {
                // 将连接设置到threadLocal中，确保事务都在同一条线程上进行
                logger.info("连接" + pooledConnection.hashCode() + "已设置为线程共享");
                THREAD_LOCAL.set(pooledConnection);
            } else {
                logger.severe("无法获得可用连接");
            }
            return pooledConnection;
        }
    };


    /**
     * 监控
     */
    private final java.lang.Object monitor = new java.lang.Object();


    private ConnectionPoolUtil() {
        createConnectionPool();
    }

    /**
     * 创建连接池，返回一个工具类对象，默认使用AbortPolicy拒绝策略
     *
     * @return {@link ConnectionPoolUtil}
     */
    public static ConnectionPoolUtil getInstance() {
        if (ValidateUtil.isNotNull(connectionPool)) {
            return connectionPool;
        }

        getInstance(new ConnectionPoolUtil.AbortPolicy());
        return connectionPool;
    }

    /**
     * 创建连接池，返回一个工具类对象，可配置拒绝策略,如果rejectedConnectionHandler为空，默认配置AbortPolicy拒绝策略
     *
     * @param rejectedConnectionHandler 拒绝连接处理程序
     * @return {@link ConnectionPoolUtil}
     */
    public static ConnectionPoolUtil getInstance(RejectedConnectionHandler rejectedConnectionHandler) {
        if (ValidateUtil.isNull(rejectedConnectionHandler)) {
            getInstance();
        }

        if (ValidateUtil.isNull(connectionPool)) {
            synchronized (ConnectionPoolUtil.class) {
                if (ValidateUtil.isNull(connectionPool)) {
                    connectionPool = new ConnectionPoolUtil();
                    handler = rejectedConnectionHandler;
                }
            }
        }
        return connectionPool;
    }

    /**
     * 得到连接
     *
     * @return {@link Connection}
     */
    public PooledConnection getConnection() {
        // 获得可用的连接
        PooledConnection poolConnection = getAvailableConnection();

        // 当获得的连接为null,让进入队列的线程在这里等待
        while (ValidateUtil.isNull(poolConnection)) {
            synchronized (monitor) {
                try {
                    // 让其等待可配置的时间，再次获取连接
                    monitor.wait(TIME_TO_WAIT);
                } catch (InterruptedException e) {
                    logger.severe(e.getMessage());
                }
            }
            int noIndex = -1;
            WRITE_LOCK.lock();
            try {
                int index = indexOf();
                // 当索引不为-1时，证明连接池集合中有不忙的连接，可以直接获取
                if (index != noIndex) {
                    // 从连接池集合中获取该不忙的连接
                    poolConnection = connections.get(index);
                    // 标志位设置为忙
                    sign.set(index, 2);
                }
            } finally {
                WRITE_LOCK.unlock();
            }
        }
        // 设置超时的时间
        poolConnection.setCheckOutTime(System.currentTimeMillis());
        return poolConnection;

    }

    /**
     * 释放资源
     *
     * @param connection 连接
     */
    public void release(Connection connection) {

        // 判断连接是否为空
        if (ValidateUtil.isNotNull(connection)) {
            // 遍历connections集合
            for (PooledConnection pooledConnection : connections) {
                // 判断封装的connection是否为空
                if (ValidateUtil.isNotNull(pooledConnection)) {
                    // 比较哈希值
                    if (pooledConnection.equals(connection)) {
                        returnConnection(pooledConnection);
                        break;
                    }
                }
            }
        }
    }

    public void closeConnectionPool() {
        WRITE_LOCK.lock();
        try {
            logger.info("开始关闭数据库连接池");
            checkConnectionPool.shutdown();
            Thread.sleep(2000);
            logger.info("检查连接线程池已关闭");
            connectionsBlockingQueue = null;
            logger.info("连接阻塞队列已清空");
            connections = null;
            logger.info("连接池已清空");
            connectionPool = null;
            logger.info("数据库连接池关闭成功");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            WRITE_LOCK.unlock();
        }
    }

    /**
     * 得到事务连接
     *
     * @return {@link Connection}
     */
    public Connection getTransactionConnection() {
        WRITE_LOCK.lock();
        try {
            // 在threadLocal中获取封装连接
            PooledConnection pooledConnection = THREAD_LOCAL.get();
            return pooledConnection.connection;
        } finally {
            WRITE_LOCK.unlock();
        }
    }

    /**
     * 返回事务连接
     */
    public void returnTransactionConnection() {
        // 从threadLocal中获取连接
        PooledConnection pooledConnection = THREAD_LOCAL.get();
        // 将该连接移出threadLocal
        THREAD_LOCAL.remove();
        // 将这个连接返回到连接池或关闭
        returnConnection(pooledConnection);
    }

    /**
     * 返回连接
     *
     * @param poolConnection 连接
     */
    private void returnConnection(PooledConnection poolConnection) {

        // 获得该连接在集合中的索引
        // 设置标志位为1
        WRITE_LOCK.lock();
        try {
            int returnIndex = connections.indexOf(poolConnection);
            sign.set(returnIndex, 1);
            // 如果标志位为1的数量大于初始化连接数，关闭该连接并把标志位设置为0
            if (notBusySize() > INIT_CONNECTIONS) {
                // 更改索引位置的连接为null
                connections.set(returnIndex, null);
                try {
                    // 关闭连接
                    poolConnection.connection.close();
                    // 标志位设置为无连接
                    sign.set(returnIndex, 0);
                } catch (SQLException e) {
                    logger.severe("连接" + poolConnection.hashCode() + "关闭失败");
                }
            }
        } finally {
            WRITE_LOCK.unlock();
        }
    }

    /**
     * 得到可用的连接
     *
     * @return {@link Connection}
     */
    private PooledConnection getAvailableConnection() {
        PooledConnection takeConnection = null;
        int noIndex = -1;
        int index;
        //如果空闲连接池不为空，直接获取连接

        // 当索引不为-1时，证明连接池集合中有不忙的连接，可以直接获取
        WRITE_LOCK.lock();
        try {
            index = indexOf();
            if (index != noIndex) {
                // 从连接池集合中获取该不忙的连接
                takeConnection = connections.get(index);
                // 标志位设置为忙
                sign.set(index, 2);
            }
        } finally {
            WRITE_LOCK.unlock();
        }

        if (index == noIndex) {
            // 创建标志位连接
            createSignConnections();
            WRITE_LOCK.lock();
            try {
                int nextIndex = indexOf();
                // 获取是否有新的不忙连接被创建
                if (nextIndex != noIndex) {
                    // 从连接池集合中获取该不忙的连接
                    takeConnection = connections.get(nextIndex);
                    // 标志位设置为忙
                    sign.set(nextIndex, 2);
                }
            } finally {
                WRITE_LOCK.unlock();
            }
        }

        return takeConnection;
    }

    /**
     * 创建连接池，为idleConnections和activeConnections创建CopyOnWriteArrayList对象
     */
    private void createConnectionPool() {
        logger.info("连接池初始化连接数：" + INIT_CONNECTIONS);
        logger.info("连接池最大连接数：" + MAX_CONNECTIONS);
        logger.info("连接池最大平时连接数：" + MAX_USUAL_CONNECTIONS);
        logger.info("连接超时时间：" + MAX_CONNECTION_TIME + "mills");
        logger.info("连接测试表为" + TEST_TABLE);
        logger.info("测试连接定时任务每次间隔时间" + TEST_CONNECTIONS_TIME);
        checkConnectionPool = new ScheduledThreadPoolExecutor(3, new CheckConnectionThreadFactory("定时任务"));
        // 为连接池设置创建时间
        ConnectionsTotal.setCreatedTime();
        logger.info("检查连接线程池创建成功，核心线程为2");
        connections = new CopyOnWriteArrayList<>();
        // 在集合中添加12个null
        for (int i = 0; i < MAX_CONNECTIONS; i++) {
            connections.add(null);
        }
        // 十个正常连接，两个救急连接
        sign = new AtomicIntegerArray(MAX_CONNECTIONS);
        connectionsBlockingQueue = new ArrayBlockingQueue<>(CONNECTIONS_BLOCKING_QUEUE_SIZE);
        logger.info("数据库连接池初始化成功！");
        for (int i = 0; i < INIT_CONNECTIONS; i++) {
            createSignConnections();
        }
        getSchedule();
    }

    /**
     * 创建新连接
     *
     * @return {@link Connection}
     */
    private PooledConnection newConnection() {
        PooledConnection poolConnection = null;
        WRITE_LOCK.lock();
        try {
            // 创建一个新连接
            poolConnection = new PooledConnection(DriverManager.getConnection(URL, USER_NAME, PASSWORD));
            ConnectionsTotal.createdConnectionsIncrement();
            // 当连接池和队列都无法满足时
            if (ValidateUtil.getNotNullSize(connections) == MAX_CONNECTIONS && ValidateUtil.judgeListEqualSize(connectionsBlockingQueue, CONNECTIONS_BLOCKING_QUEUE_SIZE) && notBusySize() == 0) {
                rejectConnection(poolConnection);
            }
        } catch (SQLException e) {
            logger.severe("创建连接失败" + e.getCause());
        } finally {
            WRITE_LOCK.unlock();
        }
        return poolConnection;
    }

    /**
     * 创建标志连接
     */
    private void createSignConnections() {

        //  创建队列连接或从队列拿出

        if (createConnectionOrPollByQueue()) {
            return;
        }
        // 当队列小于最大容量时添加
        if (tryPut()) {
            return;
        }
        // 当连接池集合大小小于最大连接数且队列长度大于最大容量时
        createEmergencyConnection();


    }

    /**
     * 拒绝策略
     */
    private void rejectConnection(PooledConnection pooledConnection) {
        try {
            handler.rejectedConnection(pooledConnection, connectionPool);
        } catch (Exception e) {
            logger.severe(e.getMessage());
        }
    }

    /**
     * 创建紧急连接
     */
    private void createEmergencyConnection() {
        if (ValidateUtil.getNotNullSize(connections) < MAX_CONNECTIONS && ValidateUtil.judgeListEqualSize(connectionsBlockingQueue, CONNECTIONS_BLOCKING_QUEUE_SIZE)) {
            // 创建救急线程
            PooledConnection pooledConnection;
            PooledConnection emergencyConnection;
            WRITE_LOCK.lock();
            try {
                pooledConnection = connectionsBlockingQueue.poll();
                emergencyConnection = newConnection();
            } finally {
                WRITE_LOCK.unlock();
            }

            try {
                connectionsBlockingQueue.offer(emergencyConnection, MAX_CONNECTION_TIME, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            WRITE_LOCK.lock();
            try {
                // 获取连接池集合位置为null的索引
                int setIndex = connections.indexOf(null);
                if (setIndex != -1) {
                    connections.set(setIndex, pooledConnection);
                    sign.set(setIndex, 1);
                }
            } finally {
                WRITE_LOCK.unlock();
            }
        }
    }

    /**
     * 创建队列连接或从队列拿出
     *
     * @return boolean
     */
    private boolean createConnectionOrPollByQueue() {
        PooledConnection pooledConnection;
        // 判断连接池的大小是否小于最大平时连接数
        WRITE_LOCK.lock();
        try {
            if (ValidateUtil.getNotNullSize(connections) < MAX_USUAL_CONNECTIONS) {
                if (ValidateUtil.judgeListMoreSize(connectionsBlockingQueue, 0)) {
                    // 取出队列内容
                    pooledConnection = connectionsBlockingQueue.poll();
                } else {
                    pooledConnection = newConnection();
                }
                // 获取连接池集合位置为null的索引

                int index = connections.indexOf(null);
                connections.set(index, pooledConnection);
                // 数组的索引对应连接集合的索引，当值为1时，代表该连接不忙，值为2时，代表该连接忙
                sign.getAndSet(index, 1);
                return true;
            }
        } finally {
            WRITE_LOCK.unlock();
        }

        return false;
    }

    /**
     * 试着把封装好的连接放进阻塞队列中暂存
     *
     * @return boolean
     */
    private boolean tryPut() {
        PooledConnection pooledConnection = newConnection();
        WRITE_LOCK.lock();
        try {
            // 当队列小于队列容量时候可以往队列添加连接
            if (ValidateUtil.judgeListLessSize(connectionsBlockingQueue, CONNECTIONS_BLOCKING_QUEUE_SIZE)) {
                // 添加进队列，超时时间可配置
                connectionsBlockingQueue.offer(pooledConnection, QUEUE_TIME_OUT, TimeUnit.MILLISECONDS);
                return true;
            }
        } catch (InterruptedException e) {
            logger.severe("暂存队列失败！");
        } finally {
            WRITE_LOCK.unlock();
        }
        return false;
    }

    /**
     * 测试连接是否可用
     */
    private void testConnections() {
        try {
            // 连接池不为空且连接池集合大小大于0
            if (ValidateUtil.isNotNull(connections) && ValidateUtil.getNotNullSize(connections) > 0) {
                // 遍历集合
                for (PooledConnection connection : connections) {
                    // 判断是否为空
                    if (ValidateUtil.isNotNull(connection)) {
                        testConnectionIsAvailableAndCheckOut(connection);
                    }
                }
            }
        } catch (Exception e) {
            logger.severe("检测异常 " + e.getCause());
        }
    }

    /**
     * 测试连接是否可用和超时
     *
     * @param pooledConnection 合用的连接
     */
    private void testConnectionIsAvailableAndCheckOut(PooledConnection pooledConnection) {
        // 获得该连接在集合内的索引
        int index = connections.indexOf(pooledConnection);
        int signBusy = sign.get(index);
        int isBusy = 2;
        WRITE_LOCK.lock();
        try {
            // 当连接标志位为2时
            if (signBusy == isBusy) {
                // 判断在忙连接是否超时
                boolean checkOutFlag = System.currentTimeMillis() - pooledConnection.getCheckOutTime() > MAX_CONNECTION_TIME;
                if (checkOutFlag) {
                    // 检测到超时将连接就移出连接池，关闭该连接
                    PooledConnection setConnection = connections.set(index, null);
                    // 连接池进行等待，等待该连接任务进行，超过时间直接关闭！
                    connectionPool.wait(TIME_TO_WAIT);
                    // 判断连接是否还在忙
                    if (sign.get(index) == isBusy) {
                        setConnection.connection.close();
                        logger.info("系统自动检测到连接" + pooledConnection.hashCode() + "超时，已将该连接关闭");
                        ConnectionsTotal.closedConnectionsIncrement();
                        ConnectionsTotal.checkOutConnectionsIncrement();
                    }
                }
            } else {
                // 测试连接是否可用
                testConnectionAvailable(pooledConnection);
            }
        } catch (SQLException | InterruptedException e) {
            logger.severe("测试功能异常！");
        } finally {
            WRITE_LOCK.unlock();
        }
    }

    /**
     * 测试连接是否可用
     *
     * @param pooledConnection 合用的连接
     */
    private void testConnectionAvailable(PooledConnection pooledConnection) {

        String sql = "SELECT COUNT(*) FROM" + TEST_TABLE;
        Connection connection = pooledConnection.connection;
        try (
                PreparedStatement preparedStatement = connection.prepareStatement(sql);
                ResultSet resultSet = preparedStatement.executeQuery()) {
            if (resultSet.next()) {
                logger.info("系统自动检测连接" + pooledConnection.hashCode() + "可用");
                returnConnection(pooledConnection);
            } else {
                throw new SQLException("该连接不可用");
            }
        } catch (Exception e) {
            logger.severe("连接" + pooledConnection.hashCode() + "出现异常" + e.getCause());
            // 不可用的连接要进行关闭
            ConnectionsTotal.checkedUnavailableConnectionsIncrement();
            int index = connections.indexOf(pooledConnection);
            PooledConnection setConnection = connections.set(index, null);
            sign.set(index, 0);
            try {
                setConnection.connection.close();
            } catch (SQLException e1) {
                logger.severe("连接关闭异常");
            }
            ConnectionsTotal.closedConnectionsIncrement();
        }
    }


    /**
     * 进行定时任务
     */
    private void getSchedule() {

        int defaultTestConnectionsTime = 10000;
        int time;
        // 避免定时任务的过多执行，强制要求配置文件定义的超时时间大小要大于默认值，否则以默认值进行
        if (TEST_CONNECTIONS_TIME >= defaultTestConnectionsTime) {
            time = TEST_CONNECTIONS_TIME;
        } else {
            time = defaultTestConnectionsTime;
        }
        // 定时检测任务
        checkConnectionPool.scheduleAtFixedRate(() -> {
            try {
                testConnections();
            } catch (Exception e) {
                logger.severe("定时检测任务异常" + e.getCause());
            }
        }, 10000, time, TimeUnit.MILLISECONDS);

        // 定时统计任务
        checkConnectionPool.scheduleAtFixedRate(() -> {
            try {
                ConnectionsTotal.poolTotal(logger);
            } catch (Exception e) {
                logger.severe("定时统计任务异常" + e.getCause());
            }
        }, 20000, 10000, TimeUnit.MILLISECONDS);
    }


    /**
     * 获得连接池不忙连接的数量
     *
     * @return int
     */
    private int notBusySize() {
        int count = 0;
        for (int i = 0; i < MAX_CONNECTIONS; i++) {
            if (sign.get(i) == 1) {
                count++;
            }
        }
        return count;
    }


    /**
     * 获得不忙连接的索引
     *
     * @return int
     */
    private int indexOf() {
        for (int i = 0; i < MAX_CONNECTIONS; i++) {
            int notBusy = 1;
            if (notBusy == sign.get(i)) {
                return i;
            }
        }
        return -1;
    }

    /**
     * 拒绝策略在拒绝获取连接时，会直接抛出一个类型为 RejectedConnectionException 的 RuntimeException
     *
     * @author HHLJ
     * @date 2022/04/13
     */
    public static class AbortPolicy implements RejectedConnectionHandler {
        public AbortPolicy() {
        }

        @Override
        public void rejectedConnection(PooledConnection pooledConnection, ConnectionPoolUtil connectionPool) throws SQLException {
            pooledConnection.connection.close();
            throw new RejectConnectionException("连接" + pooledConnection.hashCode() +
                    " 被连接池 " +
                    connectionPool.toString() +
                    "拒绝");
        }
    }

    /**
     * 丢弃新创建的连接，但是不抛出异常。
     *
     * @author HHLJ
     * @date 2022/04/13
     */
    public static class DiscardPolicy implements RejectedConnectionHandler {

        public DiscardPolicy() {
        }

        @Override
        public void rejectedConnection(PooledConnection pooledConnection, ConnectionPoolUtil connectionPool) throws SQLException {
            pooledConnection.connection.close();
        }
    }

    /**
     * 丢弃队列中最老的连接，放入新创建的连接
     *
     * @author HHLJ
     * @date 2022/04/13
     */
    public static class DiscardOldestPolicy implements RejectedConnectionHandler {
        public DiscardOldestPolicy() {
        }

        @Override
        public void rejectedConnection(PooledConnection connection, ConnectionPoolUtil connectionPool) throws InterruptedException {
            connectionPool.connectionsBlockingQueue.poll();
            connectionPool.connectionsBlockingQueue.offer(connection, QUEUE_TIME_OUT, TimeUnit.MILLISECONDS);
        }
    }

    /**
     * 什么都不干，死等呗
     *
     * @author HHLJ
     * @date 2022/04/13
     */
    public static class StayWaitPolicy implements RejectedConnectionHandler {
        public StayWaitPolicy() {
        }

        @Override
        public void rejectedConnection(PooledConnection connection, ConnectionPoolUtil connectionPool) {

        }
    }
}
